package hivetohbase_java;

import util.HbaseUtil;
import hivetohbase_scala.HiveUtils;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;

import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

import org.apache.spark.sql.types.StructField;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import scala.Tuple2;
import scala.Tuple3;


import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;


public class HbaseSpark {

    private Table table;

    @Before
    public void getAdmin() {
        table = HbaseUtil.getTable("hive_hbase_1:t_user");
    }

    @After
    public void close() {
        HbaseUtil.closeTable(table);
    }


    /**
     * 向table中添加数据：put
     * 批量增加
     */
    @Test
    public void putData_Multi()  throws IOException {

        String hadoopUrl = "hdfs://192.168.1.171:9000";
        String hiveUrl = "hdfs://192.168.1.171:9000/user/hive/warehouse";
        String hbaseUrl = "192.168.1.171:2181";
        String sysUser = "work";
        String hBaseTableName = "hive_hbase_1:bol_dw_main_material_place_order";
        String tmpDir = "/tmp/test-hbase" ;//临时文件保存位置，在hdfs上

        String hBaseColumnFamily = "cf1";


        /***********************获取hive_scala配置********************************/
//        HiveSpark hiveSpark = new HiveSpark();
//        Dataset<Row> columnsInfo = hiveSpark.getColumnsInfo2();

        /***********************获取hive_java配置********************************/

        HiveSpark hiveSpark = new HiveSpark();
        Dataset<Row> rowDataset = hiveSpark.getColumnsInfo();
//        rowDataset.show();

        for (String column : rowDataset.columns()) {
            System.out.println(column);
        }


        JavaRDD<Tuple2<String, Tuple3<String, String, String>>> dataRDD = rowDataset.javaRDD().flatMap(new FlatMapFunction<Row, Tuple2<String, Tuple3<String, String, String>>>() {
            @Override
            public Iterator<Tuple2<String, Tuple3<String, String, String>>> call(Row row)  throws Exception {
                ArrayList<Tuple2<String, Tuple3<String, String, String>>> arr = new ArrayList<>();

                String rowKey = row.getAs("rowkey".toLowerCase());
                scala.collection.Iterator<StructField> iterator =  row.schema().iterator();


                while (iterator.hasNext()) {
                    StructField field = iterator.next();
                    String columnFamily = "cf2";
                    String columnName = field.name();
                    String columnValue = row.getAs(field.name());
                    arr.add(new Tuple2<>(rowKey, new Tuple3<>(columnFamily, columnName, columnValue)));
                }

                return arr.iterator();
            }
        });
/**
 * new Function<Tuple2<String, Tuple3<String, String, String>>, Tuple3<String, String, String>>() {
 *             @Override
 *             public Tuple3<String, String, String> call(Tuple2<String, Tuple3<String, String, String>> x) throws Exception {
 *                 return new Tuple3<>(x._1, x._2._1(), x._2._2());
 *             }}
 */

        JavaRDD<Tuple2<ImmutableBytesWritable, KeyValue>> resRDD = dataRDD.filter(new Function<Tuple2<String, Tuple3<String, String, String>>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, Tuple3<String, String, String>> value) throws Exception {

                return value._1 != null;
            }
        }).sortBy(new Function<Tuple2<String, Tuple3<String, String, String>>, String>() {
            public String call(Tuple2<String, Tuple3<String, String, String>> v1) throws Exception {

                return v1._1+v1._2._1()+v1._2._2();
            }
        }).map(new Function<Tuple2<String, Tuple3<String, String, String>>, Tuple2<ImmutableBytesWritable, KeyValue>>() {
            @Override
            public Tuple2<ImmutableBytesWritable, KeyValue> call(Tuple2<String, Tuple3<String, String, String>> v3) throws Exception {
                byte[] rowkey = Bytes.toBytes(v3._1);
                byte[] family = Bytes.toBytes(v3._2._1());
                byte[] column = Bytes.toBytes(v3._2._2());
                byte[] value = Bytes.toBytes(v3._2._3());
                return new Tuple2<>(new ImmutableBytesWritable(rowkey), new KeyValue(rowkey, family, column, value));

            }
        });

//        Iterator<Tuple2<ImmutableBytesWritable, KeyValue>> iterator = resRDD.collect().iterator();
//
//        while (iterator.hasNext()){
//            System.out.println(iterator.next()._1+" ,"+iterator.next()._2);
//        }

//        hivetohbase_scala.HbaseSpark.write2Hbase(hbaseUrl,resRDD.rdd(),hBaseColumnFamily, TableName.valueOf(hBaseTableName+"1111"), tmpDir, hadoopUrl, sysUser);
//
//        HiveUtils.hiveConnect(hiveUrl, sysUser).close();



    }

}
